package com.shujia.flink.sink

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

import java.sql.{Connection, DriverManager, PreparedStatement}
object Demo03MySQLSink {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val lineDS: DataStream[String] = env.socketTextStream("master", 8888)

    val countDS: DataStream[(String, Int)] = lineDS.flatMap(_.split(","))
      .map((_, 1))
      .keyBy(_._1)
      .sum(1)



    countDS.addSink(new MysqlSink)
    env.execute()

  }

}

/**
 * SinkFunction:自定义sink
 * RichSinkFunction: 多个open和close方法，用于打开和关闭网络连接
 */
class  MysqlSink extends   RichSinkFunction[(String,Int)]{
  var con: Connection = _
  var stat: PreparedStatement = _

  override def open(parameters: Configuration): Unit = {


    //1.加载驱动
    Class.forName("com.mysql.jdbc.Driver")

    //2.创建连接
    val con: Connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/student", "root", "")

    //3、编写插入数据的sql
    //replace into ： 如果不存在插入，如果存在就更新
    stat = con.prepareStatement("replace into WordCount(words,num) values(?,?)")
  }

  //在task回收的时候执行
  override def close(): Unit = {
    //关闭连接
    stat.close()
    con.close()
  }

  override def invoke(value: (String, Int), context: SinkFunction.Context): Unit = {
    /**
     * 将数据保存到mysql中  -jdbc
     *
     */
    //4、指定sql的参数
    stat.setString(1, value._1)
    stat.setLong(2, value._2)
    //5、执行插入
    stat.execute()

  }
}
